import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 滑动窗口计算
 * 通过socket模拟产生单词数据
 * flink对数据进行统计计算
 */
public class SocketWindowWordCount {

    public static void main(String[] args) throws Exception {
         Logger log = LoggerFactory.getLogger("SocketWindowWordCount");
        //获取socket的端口号
        int port;
        try {
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            port = parameterTool.getInt("port");
        }catch (Exception e){
            System.out.println("No port set. use default port 9000");
            port = 9000;
        }
        log.info("hello5555");
        // 获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String hostname = "192.168.0.105";
        String delimiter = "\n";
        DataStreamSource<String> text  = env.socketTextStream(hostname, port, delimiter);
        // 跟spark一样，使用flatmap算子来操作
        // 输入数据为string类型，输出为自定义的WordWithCount类型对象
        DataStream<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction<String, WordWithCount>() {
            public void flatMap(String value, Collector<WordWithCount> out) {
                String[] splits = value.split(" ");
                for (String word : splits) {
                    out.collect(new WordWithCount(word, 1L));
                }
            }
        }).keyBy("word")
                // 指定时间窗口大小为10秒，指定时间间隔为5秒
                .timeWindow(Time.seconds(10), Time.seconds(5))
                // 每隔1秒统计前2秒的数据
                .sum("count");

        //把数据打印到控制台并且设置并行度
        windowCounts.print().setParallelism(1);
        env.execute("Socket window count");
    }

    public static class WordWithCount {
        public String word;
        public Long count;
        public  WordWithCount(){}
        public WordWithCount(String word,long count){
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return "WordWithCount{" +
                    "word='" + word + '\'' +
                    ", count=" + count +
                    '}';
        }
    }

}